package com.gj;

import com.gj.feign.TransactionMessageFeign;
import com.gj.model.TransactionMessage;
import org.apache.commons.lang.time.DateFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**
 * author:guijie
 * date:${date}
 * description:TODO
 */
@Service
public class ProcessMessageTask {

    private static final Logger logger = LoggerFactory.getLogger(ProcessMessageTask.class);

    @Autowired
    private TransactionMessageFeign feign;    //注入消息持久化系统
    @Autowired
    private KafkaTemplate<String,TransactionMessage> kafkaTemplate;

    private ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);   //创建线程池
    private Semaphore semaphore = new Semaphore(20);   //规定线程数为20

    public void start(){
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    //TODO 这里需要实现分布式锁
                    try {
                        System.out.println("开始发送消息:" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
                        int sleepTime = process();
                        if (sleepTime>0){
                            Thread.sleep(10000);
                        }
                    }catch (Exception e){
                        logger.error("",e);
                    }finally {
                        //TODO:这里释放分布式锁
                    }
                }
            }
        });
        thread.start();
    }

    private int process() throws Exception {
        int sleepTime = 10000;	//默认执行完之后等等10秒
        List<TransactionMessage> messageList = feign.findByWatingMessage();   //查询等待消费的消息
        if (messageList.size() > 5000){  //如果待消费消息超过5000条则直接执行不阻塞
            sleepTime=0;
        }
        final CountDownLatch latch = new CountDownLatch(messageList.size());      //等待其他线程执行完才执行主线程
        for (TransactionMessage message: messageList){
            semaphore.acquire();
            fixedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        doProcess(message);
                    }catch (Exception e){
                        logger.error("",e);
                    }finally {
                        semaphore.release();
                        latch.countDown();
                    }
                }
            });
        }
        latch.await();
        return sleepTime;
    }

    private void doProcess(TransactionMessage message) {
        //检查该消息是否满足死亡状态,如果确认为死亡状态就标记为死亡消息
        if (message.getSendCount() > message.getDieCount()){
            feign.confirmDieMessage(message.getId());
            return;
        }

        //距离上次发送时间超过一分钟才继续发送
        long currentTime = System.currentTimeMillis();
        long sendTime = 0;
        if (message.getSendDate()!=null){
            sendTime = message.getSendDate().getTime();
        }
        //距离上次发送时间大于1分钟才发送
        if (currentTime - sendTime>60000){
            System.out.println("发送具体消息：" + message.getId());

            //向MQ发送消息
            kafkaTemplate.send("transaction", message);
            //累加发送次数以及跟新发送时间
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String date = format.format(new Date());
            feign.incrSendCount(message.getId(), date);
        }
    }
}
